package com.hskn.hss.config;

import com.alibaba.fastjson.JSONObject;
import com.hskn.hss.module.kfkanalyse.service.IkfkAnalyseService;
import com.hskn.hss.module.kfkdeal.service.impl.KfkHandleServiceImpl;
import com.hskn.hss.utils.KafkaUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.Date;
import java.util.Map;

/**
 * @author lny
 * createTime 2021-09-09
 **/
@Order(value = 2)
@Slf4j
@Component
public class KfkCarLocationRunner implements CommandLineRunner {

    @Autowired
    private KfkHandleServiceImpl kfkHandleService;
    @Autowired
    IkfkAnalyseService ikfkAnalyseService;

    @Value("${kafka.car_location_topic}")
    private String car_location_topic;

    @Value("${kafka.car_location_group}")
    private String car_location_group;

    @Value("${kafka.kafka_receive:true}")
    private Boolean kafka_receive;

    @Override
    public void run(String... args) {
        if (!kafka_receive) {
            return;
        }
        new Thread(() -> {
            KafkaConsumer consumer = KafkaUtils.getConsumer(car_location_group);
            consumer.subscribe(Arrays.asList(car_location_topic));
            while (true) {
                receiveAndHandle(consumer);
            }
        }).start();
    }

    public void receiveAndHandle(KafkaConsumer consumer) {
        //拉取kafka数据
        log.info("====================kfk定时读取任务开始-车辆======================================");
        try {
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            log.info("========KfkCarLocationRunner拉取条数：{}", records.count());
            for (ConsumerRecord<Object, Object> record : records) {
                try {
                    Map value = JSONObject.parseObject(String.valueOf(record.value()), Map.class);
                    Map content = JSONObject.parseObject(String.valueOf(value.get("content")), Map.class);
                    Date kfkTime = new Date(record.timestamp());

                    if (value.get("deviceType").equals("CAR_LOCATION") && content.get("type").equals("LOCATION_REPORT")) { //车载定位终端
                        long carStartTime = System.currentTimeMillis();
                        kfkHandleService.carLocationHandle(value, kfkTime);
                        long carEndTime = System.currentTimeMillis();
                        log.info("=====车载定位终端=执行时间(毫秒)：{}", carEndTime - carStartTime);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
            log.info("====================kfk定时读取任务结束-车辆======================================");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            consumer.commitAsync();
        }
    }
}